package com.acme.bankapp.service.bank.listener.email;

import com.acme.bankapp.domain.bank.client.Client;
import com.acme.bankapp.service.bank.listener.ClientRegistrationListener;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author Alexander Kotov (kotov.alex.22@gmail.com)
 */
public class EmailService implements ClientRegistrationListener {
    transient private BankBlockingQueue<Email> queue;

    transient private final ExecutorService consumersPool;
    transient private final ExecutorService producersPool;
    private final int consumersPoolSize;
    private final int maxQueueSize;

    public EmailService(int maxQueueSize, int consumersPoolSize, int producersPoolSize) {
        this.consumersPoolSize = consumersPoolSize;
        this.maxQueueSize = maxQueueSize;
        queue = new BankBlockingQueue<Email>(maxQueueSize);
        producersPool = Executors.newFixedThreadPool(producersPoolSize);
        consumersPool = Executors.newFixedThreadPool(consumersPoolSize);
        for (int i = 0; i != consumersPoolSize; ++i) {
            consumersPool.submit(new EmailWorker(queue));
        }
    }

    @Override
    public void onClientAdded(Client client) {
        Email email = new Email(client, client.getEmailAddress());
        producersPool.submit(new EmailProducer(email, queue));
    }

    public void close() {
        queue.close();
        producersPool.shutdownNow(); // Disable new tasks from being submitted
        consumersPool.shutdown(); // Disable new tasks from being submitted
        try {
            // Wait a while for existing tasks to terminate
            if (!consumersPool.awaitTermination(100, TimeUnit.MILLISECONDS)) {
                consumersPool.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!consumersPool.awaitTermination(100, TimeUnit.MILLISECONDS))
                    System.err.println("Pool of consumers did not terminate");
            }
            if (!producersPool.awaitTermination(100, TimeUnit.MILLISECONDS)) {
                producersPool.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!producersPool.awaitTermination(100, TimeUnit.MILLISECONDS))
                    System.err.println("Pool of producers did not terminate");
            }
        } catch (InterruptedException ie) {
            // (Re-)Cancel if current thread also interrupted
            consumersPool.shutdownNow();
            producersPool.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
}
